#include <sys/file.h>
#include <sys/stat.h>
#include <unistd.h>

#include "StoreObject.h"
#include "StoreObjectIO.h"

namespace gitstore {
CObjectIOBase::CObjectIOBase()
{

}

CObjectIOBase::~CObjectIOBase()
{

}

ErrorStatus CObjectIOBase::setValue(const std::string& key, const StoreObject& value) {
    return setValue(key, value.getSerializedValue());
}

ErrorStatus CObjectIOBase::getValue(const std::string& key, StoreObject& value) {
    std::string svalue;
    auto res = this->getValue(key, svalue);
    if ( Success != res )
        return res;
    value.fromSerializedValue(std::move(svalue));
    return Success;
}

CLinuxObjectIOBase::CLinuxObjectIOBase() {

}

CLinuxObjectIOBase::~CLinuxObjectIOBase() {

}


//ErrorStatus CLinuxObjectIOBase::setValue(const std::string& key, const StoreObject& value) {
//    string file_path = this->store_path+key;
//    int fd = open(file_path.c_str(), O_WRONLY|O_TRUNC|O_CREAT, S_IRUSR);
//    if ( fd < 0 )
//        return FileSystemError;
//    flock(fd, LOCK_EX);

//    int ws = write(fd, value.getData(), value.getSize());
//    close(fd);
//    return Success;
//}

//ErrorStatus CLinuxObjectIOBase::getValue(const std::string& key, StoreObject& value) {
//    std::string content;
//    string file_path = this->store_path+key;
//    int fd = open(file_path.c_str(), O_RDONLY, S_IRUSR);
//    if ( fd < 0 )
//        return FileSystemError;
//    //flock(fd, LOCK_EX);

//    flock(fd, LOCK_SH);
//    //cout<<"lock success"<<endl;
//    struct stat st;
//    int fs = fstat(fd, &st);
//    if ( fs < 0 )
//    {
//        close(fd);
//        return FileSystemError;
//    }
//    char *buf = new char[st.st_size];
//    ssize_t rs = read(fd, buf, st.st_size);
//    if ( rs < 0 )
//    {
//        close(fd);
//        return FileSystemError;
//    }
//    if (value.getType() == StoreObjectType::Blob) {
//        value.setValue(buf, rs);
//    } else {
//        value.setValue(buf);
//    }
//    content.assign(buf, rs);
//    if ( buf )
//        delete buf;
//    //flock(fd, LOCK_UN);
//    close(fd);

//    return Success;
//}

ErrorStatus CLinuxObjectIOBase::setValue(const std::string& key, const std::string& value) {
    return Success;
}

ErrorStatus CLinuxObjectIOBase::getValue(const std::string& key, std::string& value) {
    return Success;
}

#ifdef COMPILE_CEPH
CCephObjectIO::CCephObjectIO(const std::string& cluster_name, const std::string& user_name, const std::string& config_file_path, const std::string& pool_name) {
    int ret = 0;
    /* Declare the cluster handle and required variables. */
    uint64_t flags ;

    /* Initialize the cluster handle with the "ceph" cluster name and "client.admin" user */
    {
        ret = _cluster.init2(user_name.c_str(), cluster_name.c_str(), flags);
        if (ret < 0) {
            //            std::cerr << "Couldn't initialize the cluster handle! error " << ret << std::endl;
            ret = EXIT_FAILURE;
            throw "Ceph Init Error";
        } else {
            //            std::cout << "Created a cluster handle." << std::endl;
        }
    }

    /* Read a Ceph configuration file to configure the cluster handle. */
    {
        ret = _cluster.conf_read_file(config_file_path.c_str());
        if (ret < 0) {
            //            std::cerr << "Couldn't read the Ceph configuration file! error " << ret << std::endl;
            ret = EXIT_FAILURE;
            //            throw "Ceph Conf Read Error";
        } else {
            //            std::cout << "Read the Ceph configuration file." << std::endl;
        }
    }
    /* Connect to the cluster */
    {
        ret = _cluster.connect();
        if (ret < 0) {
            //            std::cerr << "Couldn't connect to cluster! error " << ret << std::endl;
            ret = EXIT_FAILURE;
            throw "Ceph Connect Error";
        } else {
            //            std::cout << "Connected to the cluster." << std::endl;
        }
    }

    {
        ret = _cluster.ioctx_create(pool_name.c_str(), _io_ctx);
        if (ret < 0) {
            //            std::cerr << "Couldn't set up ioctx! error " << ret << std::endl;
            //exit(EXIT_FAILURE);
            throw "Ceph IO Create Error";
        } else {
            //            std::cout << "Created an ioctx for the pool." << std::endl;
        }
    }
}

CCephObjectIO::~CCephObjectIO() {
    _io_ctx.close();
    _cluster.shutdown();
}

//ErrorStatus CCephObjectIO::setValue(const std::string& key, const StoreObject& value) {
//    int ret = 0;
//    librados::bufferlist bl;
//    switch (value.getType()) {
//    case StoreObjectType::Blob:
//        bl.append(value.getData(), value.getSize());
//        break;
//    case StoreObjectType::String:
//        bl.append(value.getData());
//        break;
//    default:
//        break;
//    }

//    ret = _io_ctx.write_full(key.c_str(), bl);
//    if ( ret < 0 )
//        return DBSaveError;
//    else {
//        return Success;
//    }
//}

//ErrorStatus CCephObjectIO::getValue(const std::string& key, StoreObject& value) {
//    int ret = 0;
//    librados::bufferlist read_buf;
//    unsigned long read_len = 4194304;

//    ret = _io_ctx.stat(key, &read_len, nullptr);
//    if (ret < 0) {
//        return DBLoadError;
//    }

//    //Send read request.
//    ret = _io_ctx.read(key, read_buf, read_len, 0);
//    if (ret < 0) {
//        return DBLoadError;
//    }
//    switch (value.getType()) {
//    case String:
//        value.setValue(read_buf.to_str().c_str());
//        break;
//    case Blob:
//        value.setValue(read_buf.to_str().data(), read_len);
//        break;
//    default:
//        return DBLoadError;

//    }
//    return Success;
//}

ErrorStatus CCephObjectIO::setValue(const std::string& key, const std::string& value) {
    int ret = 0;
    librados::bufferlist bl;
    bl.append(value.c_str(), value.size());
    ret = _io_ctx.write_full(key.c_str(), bl);
    if ( ret < 0 )
        return DBSaveError;
    else {
        return Success;
    }
}

ErrorStatus CCephObjectIO::getValue(const std::string& key, std::string& value) {
    int ret = 0;
    librados::bufferlist read_buf;
    unsigned long read_len = 4194304;

    ret = _io_ctx.stat(key, &read_len, nullptr);
    if (ret < 0) {
        return DBLoadError;
    }

    //Send read request.
    ret = _io_ctx.read(key, read_buf, read_len, 0);
    if (ret < 0) {
        return DBLoadError;
    }
    value.clear();
    value.reserve(read_len);
    value.assign(read_buf.to_str());
    return Success;
}

#endif
}
